package com.sunzm.flink.datastream.scala.robotcall

import java.sql.{Connection, DriverManager, PreparedStatement, ResultSet, SQLException}

import com.alibaba.fastjson.JSONObject
import io.debezium.data.Envelope.Operation
import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.functions.source.{RichSourceFunction, SourceFunction}

/**
 *
 * 从MySQL中读取费率信息，只在程序第一次启动时运行一次
 *
 * @author Administrator
 * @version 1.0
 * @date 2021-08-05 22:37
 */
class RobotCallDataSourceMySQL extends RichSourceFunction[String]{
  private var connection: Connection = null
  private var pstmt: PreparedStatement = null
  private var rs: ResultSet = null
  private val sql: String = "SELECT * FROM call_number_type_info"


  override def open(parameters: Configuration): Unit = {
    connection = DriverManager.getConnection("jdbc:mysql://82.156.210.70:3306/test?useSSL=false",
      "root", "ABC123abc.123")

    pstmt = connection.prepareStatement(sql)
  }

  override def run(ctx: SourceFunction.SourceContext[String]): Unit = {

    rs = pstmt.executeQuery
    while (rs.next) {
      val number_type = rs.getInt("number_type")
      val sip_callin_charge = rs.getDouble("sip_callin_charge")
      val sip_callout_charge = rs.getDouble("sip_callout_charge")
      val phone_callin_charge = rs.getDouble("phone_callin_charge")
      val phone_callout_charge = rs.getDouble("phone_callout_charge")

      //输出结果
      val jSONObject = new JSONObject()
      jSONObject.put("number_type", number_type)
      jSONObject.put("sip_callin_charge", sip_callin_charge)
      jSONObject.put("sip_callout_charge", sip_callout_charge)
      jSONObject.put("phone_callin_charge", phone_callin_charge)
      jSONObject.put("phone_callout_charge", phone_callout_charge)
      jSONObject.put("opType", Operation.CREATE.toString.toUpperCase)

      ctx.collect(jSONObject.toJSONString)

      //ctx.collect(number_type, (sip_callin_charge, sip_callout_charge, phone_callin_charge, phone_callout_charge))
    }

  }

  override def cancel(): Unit = {
    closeConn
  }

  override def close(): Unit = {
    closeConn
  }

  private def closeConn(): Unit = {
    if (rs != null) try rs.close()
    catch {
      case throwables: SQLException =>
        throwables.printStackTrace()
    }
    if (pstmt != null) try pstmt.close()
    catch {
      case throwables: SQLException =>
        throwables.printStackTrace()
    }
    if (connection != null) try connection.close()
    catch {
      case throwables: SQLException =>
        throwables.printStackTrace()
    }
  }
}
